/** 
 * @Title:Curator封装类
 * @Desription:Curator封装类
 * @Company:CSN
 * @ClassName:CuratorWrapper.java
 * @Author:Justic
 * @CreateDate:2015-6-30   
 * @UpdateUser:Justic  
 * @Version:0.1 
 *    
 */

package com.justic.zk.quartz.curator.impl;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Strings;
import com.justic.zk.quartz.curator.CuratorManager;
import com.justic.zk.quartz.curator.ICuratorWrapper;
import com.justic.zk.quartz.curator.support.NodeMode;
import com.justic.zk.quartz.curator.support.NodePathCacheListener;
import com.justic.zk.quartz.support.ZkConfig;

/**
 * @ClassName: CuratorWrapper
 * @Description: Curator封装类
 * @author: Justic
 * @date: 2015-6-30
 * 
 */
public class CuratorWrapper implements ICuratorWrapper {

	private static Logger logger = LoggerFactory.getLogger(CuratorWrapper.class);

	private CuratorFramework zkClient;
	private String rootPath;
	private String nodeName;
	private NodeMode nodeMode;// leader or follower
	private CuratorManager curatorManager;

	public CuratorWrapper(ZkConfig zkConfig) {
		try {

			// initial zk client
			init(zkConfig);
			// connect
			start();
			// instance CuratorManager
			curatorManager = new CuratorManager();

		} catch (Exception e) {
			logger.error("CuratorWrapper init error:{}", e);
		}
	}

	/**
	 * 监听节点变化
	 * 
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#addPathChildrenListener()
	 */
	public void addPathChildrenListener() throws Exception {

		addPathChildrenListener(rootPath);

	}

	/**
	 * 监听节点变化
	 * 
	 * @param nodePath
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#addPathChildrenListener(java.lang.String)
	 */
	@Override
	public void addPathChildrenListener(String nodePath) throws Exception {

		// start thread moniter node state
		ExecutorService pool = Executors.newFixedThreadPool(1);
		// 监听节点变化情况
		PathChildrenCache childrenCache = new PathChildrenCache(zkClient, nodePath, true);
		childrenCache.start(StartMode.POST_INITIALIZED_EVENT);
		childrenCache.getListenable().addListener(new NodePathCacheListener(childrenCache), pool);
	}

	/**
	 * get curator object
	 * 
	 * @return
	 */
	public CuratorFramework getClient() {
		return zkClient;
	}

	/**
	 * check curator started
	 * 
	 * @return
	 */
	@SuppressWarnings("deprecation")
	public boolean isStart() {
		return zkClient.isStarted();
	}

	/**
	 * start curator
	 */
	public void start() {

		if (zkClient != null && zkClient.getState() != CuratorFrameworkState.STARTED) {
			zkClient.start();
		}
	}

	/**
	 * close curator
	 */
	public void close() {
		if (zkClient != null && zkClient.getState() == CuratorFrameworkState.STOPPED) {
			zkClient.close();
		}
	}

	/**
	 * init curator
	 * 
	 * @param zkConfig
	 */
	private void init(ZkConfig zkConfig) {
		// retry policy
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
		//
		zkClient = CuratorFrameworkFactory.newClient(zkConfig.getZkServer(),
				zkConfig.getSessionTimeout() != 0 ? zkConfig.getSessionTimeout() : SESSION_TIMEOUT_MS,
				zkConfig.getConnectTimeout() != 0 ? zkConfig.getConnectTimeout() : CONNECTION_TIMEOUT_MS, retryPolicy);
		// set namespace
		zkClient.newNamespaceAwareEnsurePath(Strings.isNullOrEmpty(zkConfig.getGroupName()) ? DEFAULT_GROUP_NAME
				: zkConfig.getGroupName());

		//
		rootPath = zkConfig.getRootPath();
		nodeName = zkConfig.getNodeName();

	}

	public String getRootPath() {
		return rootPath;
	}

	public String getNodeName() {
		return nodeName;
	}

	public boolean isLeader() {
		return NodeMode.Leader.equals(nodeMode);
	}

	public void setNodeMode(NodeMode nodeMode) {
		this.nodeMode = nodeMode;
	}

	/**
	 * 创建节点
	 * 
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#createNode()
	 */
	public void createNode() throws Exception {
		createNode(rootPath.concat("/").concat(nodeName), nodeName);
	}

	/**
	 * 创建节点
	 * 
	 * @param nodePath
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#createNode(java.lang.String)
	 */
	@Override
	public void createNode(String nodePath) throws Exception {
		createNode(nodePath, "");
	}

	/**
	 * 创建节点
	 * 
	 * @param nodePath
	 * @param value
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#createNode(java.lang.String,
	 *      java.lang.String)
	 */
	@Override
	public void createNode(String nodePath, String value) throws Exception {
		curatorManager.createNode(zkClient, nodePath, value);

	}

	/**
	 * 更新节点数据
	 * 
	 * @param nodeName
	 * @param value
	 * @return
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#updateNode(java.lang.String,
	 *      java.lang.String)
	 */
	public boolean updateNode(String nodePath, String value) throws Exception {

		return curatorManager.updateNode(zkClient, nodePath, value);

	}

	/**
	 * 删除节点
	 * 
	 * @param nodePath
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#deleteNode(java.lang.String)
	 */
	public void deleteNode(String nodePath) throws Exception {
		curatorManager.deleteNode(zkClient, nodePath);
	}

	/**
	 * 获取节点路径下所有子节点路径
	 * 
	 * @param nodePath
	 * @return
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#listChildren(java.lang.String)
	 */
	public List<String> listChildren(String nodePath) throws Exception {
		return curatorManager.listChildren(zkClient, nodePath);
	}

	/**
	 * 获取节点路径下所有子节点路径、数据
	 * 
	 * @param nodePath
	 * @return
	 * @throws Exception
	 * @see com.csair.amms2.schedule.curator.ICuratorWrapper#listchildrenDetail(java.lang.String)
	 */
	public Map<String, String> listchildrenDetail(String nodePath) throws Exception {
		return curatorManager.listChildrenDetail(zkClient, nodePath);
	}

}
